# tcp_server.py
import asyncio
import json
import uuid
from typing import List, Dict, Any, Optional

from tcp_server.tcp_clients import ClientManager
from utils.log_util import logger


class TCPServer:
    """TCP服务器 - 管理agent客户端连接"""
    
    def __init__(self, host: str = '0.0.0.0', port: int = 9000):
        self.host = host
        self.port = port
        self.server = None
        self._db_session = None
        self._redis = None
    
    def set_db_session(self, db_session):
        """设置数据库会话工厂"""
        self._db_session = db_session
    
    def set_redis(self, redis):
        """设置Redis客户端"""
        self._redis = redis
    
    async def handle_client(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
        """处理客户端连接"""
        addr = writer.get_extra_info('peername')
        ip_address = addr[0]
        port = addr[1]
        
        # 临时client_id，等待客户端第一次状态上报后更新
        temp_client_id = None
        real_client_id = None
        verified = False  # 连接验证标志
        
        logger.debug(f"新连接: {ip_address}:{port}")
        
        try:
            # ✅ 增加 StreamReader 的缓冲区限制到 10MB，避免大型响应被截断
            reader._limit = 10 * 1024 * 1024  # 10MB
            
            while True:
                # 读取客户端消息（按行读取）
                try:
                    data = await reader.readline()
                except (ConnectionResetError, BrokenPipeError, OSError) as e:
                    logger.debug(f"连接被客户端重置 {ip_address}:{port}: {type(e).__name__}")
                    break
                except asyncio.CancelledError:
                    logger.debug(f"读取被取消 {ip_address}:{port}")
                    raise
                except Exception as e:
                    logger.warning(f"读取数据异常 {ip_address}:{port}: {e}")
                    break
                
                if not data:
                    logger.debug(f"客户端 {ip_address}:{port} 正常断开")
                    break
                
                # 先尝试解码，检测HTTP请求
                try:
                    raw_data = data.decode('utf-8', errors='ignore').strip()
                except:
                    logger.warning(f"无法解码数据 from {ip_address}:{port}")
                    writer.close()
                    await writer.wait_closed()
                    break
                
                # 检测HTTP请求并拒绝
                if raw_data.startswith(('GET ', 'POST ', 'PUT ', 'DELETE ', 'HEAD ', 'OPTIONS ', 'CONNECT ')):
                    logger.debug(f"❌ 检测到HTTP请求 from {ip_address}:{port}，拒绝连接")
                    writer.close()
                    await writer.wait_closed()
                    break
                
                try:
                    message = json.loads(raw_data)
                    
                    # 第一条消息必须包含event字段，否则拒绝（验证机制）
                    if not verified:
                        if 'event' not in message or 'direction' not in message:
                            logger.debug(f"❌ 非法连接 from {ip_address}:{port}，缺少协议字段")
                            writer.close()
                            await writer.wait_closed()
                            break
                        verified = True
                        logger.info(f"✅ Agent连接验证通过: {ip_address}:{port}")
                    
                    # ✅ 移除冗余日志：心跳消息每5秒一次，不需要记录
                    # logger.debug(f"收到消息 from {ip_address}:{port}: {message}")
                    
                    # 处理消息并获取client_id（传入当前已知的client_id）
                    client_id = await self.process_client_message(
                        ip_address, port, message, reader, writer, real_client_id
                    )
                    
                    if client_id:
                        real_client_id = client_id
                        
                except json.JSONDecodeError as e:
                    # 扫描器发送的垃圾数据，使用DEBUG级别
                    logger.debug(f"无效的JSON from {ip_address}:{port}: {e}")
                    writer.close()
                    await writer.wait_closed()
                    break
                except Exception as e:
                    logger.error(f"处理消息错误 {ip_address}:{port}: {e}")
        
        except (ConnectionResetError, BrokenPipeError, OSError) as e:
            # 这些是正常的网络断开，不记录详细日志
            logger.debug(f"连接异常 {ip_address}:{port}: {type(e).__name__}")
        except asyncio.CancelledError:
            logger.debug(f"连接处理被取消 {ip_address}:{port}")
        except Exception as e:
            # 只有非预期的异常才记录详细日志
            logger.error(f"客户端连接错误 {ip_address}:{port}: {e}")
        finally:
            # 客户端断开连接
            if real_client_id:
                await ClientManager.remove_client(real_client_id, self._db_session)
                logger.info(f"Agent客户端断开: {real_client_id} ({ip_address}:{port})")
            else:
                # 未验证的连接（扫描器）使用DEBUG级别
                logger.debug(f"未验证连接断开: {ip_address}:{port}")
            try:
                writer.close()
                await writer.wait_closed()
            except:
                pass
    
    async def process_client_message(
        self, 
        ip_address: str, 
        port: int, 
        message: dict,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
        known_client_id: str = None
    ) -> str:
        """处理客户端消息
        
        Args:
            known_client_id: 已知的client_id（从之前的状态消息中获取）
        """
        event = message.get("event")
        direction = message.get("direction")
        data = message.get("data", {})
        
        # 客户端ID（优先从data中获取，否则使用已知的ID）
        client_id = data.get("client_id") if data else None
        if not client_id:
            client_id = known_client_id
        
        # 状态上报消息
        # 🍪 处理Cookie更新事件（轻量级，只更新cookie和account_nickname）
        if event == "update_cookie" and direction == "res":
            if not client_id:
                logger.warning("收到update_cookie事件但缺少client_id")
                return None
            
            # 只更新cookie相关字段到数据库
            from module_admin.service.agent_client_service import AgentClientService
            async with self._db_session() as db:
                try:
                    cookie_data = data.get("cookie")
                    if cookie_data:
                        await AgentClientService.update_client_status(
                            db, client_id, {"cookie": cookie_data}
                        )
                        await db.commit()
                        logger.info(f"✅ 已更新客户端 {client_id} 的Cookie")
                    else:
                        logger.warning(f"update_cookie事件缺少cookie数据: {client_id}")
                except Exception as e:
                    logger.error(f"更新Cookie失败: {e}")
                    await db.rollback()
            
            return client_id
        
        # 处理状态更新事件（完整状态）
        if event == "status" and direction == "res":
            # 如果还是没有client_id，生成一个
            if not client_id:
                client_id = str(uuid.uuid4())
            
            # ✅ 转换 online (bool) 为 online_status (str)，确保前端能正确识别
            if isinstance(data, dict):
                if "online" in data and "online_status" not in data:
                    # 将 online (bool) 转换为 online_status (str)
                    data["online_status"] = "online" if data.get("online") else "offline"
                elif "online" in data:
                    # 如果两者都存在，优先使用 online_status，但确保它存在
                    if "online_status" not in data or not data["online_status"]:
                        data["online_status"] = "online" if data.get("online") else "offline"
            
            # ✅ 检查是否是首次连接（新客户端）
            # 注意：需要在add_or_update_client之前检查，因为add_or_update_client会将client_id添加到clients中
            # ClientManager 已在文件顶部导入，无需重复导入
            is_new_client = client_id not in ClientManager.clients
            
            # 注册或更新客户端
            await ClientManager.add_or_update_client(
                client_id=client_id,
                ip_address=ip_address,
                port=port,
                reader=reader,
                writer=writer,
                status_data=data,
                db_session=self._db_session
            )
            
            # ⚠️ 不再发送黑名单给Agent（黑名单过滤已统一在控制端进行）
            # 如果未来需要恢复Agent端过滤，可以取消注释以下代码
            # if is_new_client:
            #     try:
            #         async with self._db_session() as db:
            #             from module_admin.service.blacklist_service import BlacklistService
            #             ...
            #     except Exception as e:
            #         logger.warning(f"发送黑名单给Agent失败: {e}")
            
            # ✅ 从数据库读取client_type并合并到状态数据中（前端需要此字段判断是否为下单客户端）
            try:
                async with self._db_session() as db:
                    from module_admin.dao.agent_client_dao import AgentClientDao
                    db_client = await AgentClientDao.get_client_by_client_id(db, client_id)
                    if db_client and db_client.client_type:
                        # 合并数据库中的client_type到状态数据
                        if isinstance(data, dict):
                            # 只在client_type真正变化时记录日志（避免心跳时频繁刷屏）
                            # Agent发送的状态数据中没有client_type字段，所以每次都是None，不需要记录
                            old_client_type = data.get("client_type")
                            new_client_type = db_client.client_type
                            # 只有当新旧值不同且旧值不是None时才记录（避免每次心跳都记录 None -> order）
                            if old_client_type is not None and old_client_type != new_client_type:
                                logger.debug(f"[状态更新] client_type已更新: {old_client_type} -> {new_client_type}")
                            data["client_type"] = new_client_type
            except Exception as e:
                # 只在首次错误时记录，避免重复刷屏
                pass
            
            # ✅ 优化：只在状态变化时才转发到WebSocket（减少前端推送频率）
            await ClientManager.broadcast_status_update(client_id, data)
            
            return client_id
        
        # 其他响应消息（登录、搜索、下单等）
        elif direction == "res":
            if client_id:
                # 更新心跳
                await ClientManager.update_heartbeat(client_id, self._db_session)
                
                # 特殊处理：搜索结果需要过滤和缓存
                if event == "search" and message.get("success"):
                    # Agent返回的格式：data.items（搜索成功时）或 data.results
                    # 兼容两种格式
                    search_data = data.get("items") or data.get("results") or data.get("data", {}).get("items") or data.get("data", {}).get("results")
                    keyword = data.get("keyword") or data.get("data", {}).get("keyword")
                    
                    if search_data and keyword:
                        logger.info(f"[搜索] {client_id} 关键词={keyword}, 数量={len(search_data) if isinstance(search_data, list) else 0}")
                        if not isinstance(search_data, list):
                            logger.warning(f"[搜索] {client_id} search_data不是列表类型: {type(search_data)}")
                            search_data = []
                        
                        try:
                            from module_admin.service.search_cache_service import SearchCacheService
                            
                            # ✅ 获取user_id（用于缓存隔离和WebSocket推送）
                            # 优先级：1. 从搜索响应data中获取（如果Agent返回了） 2. 从client_user_map缓存获取
                            user_id = data.get("user_id") or data.get("data", {}).get("user_id")
                            if not user_id:
                                user_id = ClientManager.get_user_id_by_client_id(client_id)
                            
                            if not user_id:
                                logger.warning(
                                    f"[搜索] ⚠️ {client_id} 无法获取user_id, keyword={keyword}"
                                )
                            
                            # 处理搜索结果（黑名单过滤 + 新商品检测 + 缓存更新）
                            # 注意：需要使用数据库会话上下文管理器
                            async with self._db_session() as db:
                                processed_result = await SearchCacheService.process_search_results(
                                    db=db,
                                    client_id=client_id,
                                    keyword=keyword,
                                    raw_results=search_data if isinstance(search_data, list) else [],
                                    user_id=user_id  # ✅ 传递user_id用于缓存隔离
                                )
                                
                                # ✅ 确保processed_result中包含user_id（用于WebSocket推送时过滤）
                                if user_id and isinstance(processed_result, dict):
                                    processed_result['user_id'] = user_id
                                
                                # 替换原始搜索结果
                                message['data'] = processed_result
                                message['processed'] = True
                                
                                # 只在有新商品时详细记录
                                new_count = processed_result.get('new_count', 0)
                                if new_count > 0:
                                    logger.info(f"[搜索] {keyword} 处理完成: 原始={processed_result.get('total', 0)}, 过滤后={processed_result.get('filtered', 0)}, 新商品={new_count}")
                                else:
                                    logger.debug(f"[搜索] {keyword} 处理完成: 原始={processed_result.get('total', 0)}, 过滤后={processed_result.get('filtered', 0)}, 无新商品")
                                
                                # 如果启用了自动下单，且有新商品，选择下单客户端执行下单
                                new_items = processed_result.get('new_items', [])
                                
                                if new_items:
                                    # ✅ 从搜索调度器配置中获取user_id（用于过滤下单客户端）
                                    from module_task.search_scheduler import scheduler_manager
                                    # 如果提供了user_id，使用对应用户的调度器配置
                                    scheduler = None
                                    if user_id:
                                        scheduler = await scheduler_manager.get_scheduler(user_id)
                                    else:
                                        # 兼容旧代码：使用全局单例
                                        from module_task.search_scheduler import search_scheduler
                                        scheduler = search_scheduler
                                    
                                    cfg = scheduler.get_config()
                                    if cfg and cfg.enable_auto_order:
                                        # ✅ 使用轮次计数判断是否是本轮任务的第一次搜索（而不是max_publish_time）
                                        round_count = scheduler.get_round_count()
                                        is_first_round = round_count == 1
                                        
                                        # ✅ 调试日志：检查配置和判断结果
                                        logger.info(f"[自动下单判断] keyword={keyword}, user_id={user_id}, round_count={round_count}, is_first_round={is_first_round}, skip_first_search={cfg.skip_first_search}, new_items_count={len(new_items)}")
                                        
                                        if is_first_round and cfg.skip_first_search:
                                            logger.info(f"✅ 首轮搜索不自动下单：keyword={keyword}, round_count={round_count}, skip_first_search=True, 跳过自动下单")
                                        else:
                                            # 异步执行自动下单（不阻塞搜索响应）
                                            logger.info(f"📦 开始自动下单：keyword={keyword}, new_items_count={len(new_items)}, round_count={round_count}, skip_first_search={cfg.skip_first_search}")
                                            asyncio.create_task(self._handle_auto_order(
                                                new_items=new_items,
                                                keyword=keyword,
                                                order_price_limit=cfg.order_price,
                                                db_session=self._db_session,
                                                user_id=user_id  # ✅ 传递user_id用于过滤下单客户端
                                            ))
                        except Exception as e:
                            logger.error(f"处理搜索结果失败: {e}", exc_info=True)
                            # ⚠️ 处理失败时，确保message['data']仍然包含原始数据，前端可以自行处理
                            # 如果message['data']已经被替换，需要恢复原始数据
                            if not message.get('processed'):
                                # 如果处理失败，确保至少转发原始items数据，前端能显示
                                logger.warning(f"[搜索处理] 处理失败，保持原始数据格式，确保前端能收到items")
                                # message['data']保持为原始格式（包含items）
                    else:
                        if not search_data:
                            logger.warning(f"[搜索处理] search_data为空, data.keys()={list(data.keys()) if isinstance(data, dict) else 'N/A'}, data完整结构={json.dumps(data, ensure_ascii=False, default=str)[:200] if isinstance(data, dict) else 'N/A'}")
                        if not keyword:
                            logger.warning(f"[搜索处理] keyword为空, data完整结构={json.dumps(data, ensure_ascii=False, default=str)[:200] if isinstance(data, dict) else 'N/A'}")
                        # ⚠️ 即使search_data或keyword为空，仍然转发原始消息，前端可以自行判断和处理
                        logger.info(f"[搜索处理] search_data或keyword为空，转发原始消息到前端")
                
                # ✅ 特殊处理：批量历史消息（Agent批量推送）
                elif event == "batch_history_messages" and direction == "res":
                    try:
                        # 🔍 调试：打印完整消息（限制1000字符）
                        logger.info(f"[批量历史消息] 🔍 完整message: {json.dumps(message, ensure_ascii=False)[:1000]}")
                        
                        # 🔍 调试：打印原始数据结构
                        logger.info(f"[批量历史消息] 🔍 原始message keys: {list(message.keys())}")
                        logger.info(f"[批量历史消息] 🔍 data keys: {list(data.keys()) if isinstance(data, dict) else 'N/A'}")
                        logger.info(f"[批量历史消息] 🔍 message.messages存在: {'messages' in message}")
                        logger.info(f"[批量历史消息] 🔍 data.messages存在: {'messages' in data if isinstance(data, dict) else False}")
                        
                        # ✅ 批量消息数据可能在顶层，也可能在data中
                        messages = message.get('messages') or data.get('messages', [])
                        count = message.get('count') or data.get('count', 0)
                        
                        # ✅ 获取client_id（可能在顶层或data中）
                        if not client_id:
                            client_id = message.get('client_id')
                        
                        logger.info(f"[批量历史消息] 收到来自 {client_id} 的 {count} 条历史消息")
                        
                        from module_admin.service.seller_message_service import SellerMessageService
                        from sqlalchemy.exc import IntegrityError  # ✅ 使用SQLAlchemy的异常类型
                        
                        success_count = 0
                        error_count = 0
                        duplicate_count = 0
                        
                        for msg_data in messages:
                            try:
                                # ✅ 每条消息使用独立的事务，避免一条失败影响整批
                                async with self._db_session() as db:
                                    msg_data['client_id'] = client_id
                                    msg_data['is_history_sync'] = True
                                    
                                    db_message = await SellerMessageService.create_message(db, msg_data)
                                    if db_message:
                                        await db.commit()
                                        success_count += 1
                                    else:
                                        error_count += 1
                                        
                            except IntegrityError as ie:
                                # ✅ 捕获重复键错误，跳过该消息
                                if '1062' in str(ie) or 'Duplicate entry' in str(ie):
                                    duplicate_count += 1
                                    logger.debug(f"[批量历史消息] 跳过重复消息: message_id={msg_data.get('message_id')}")
                                else:
                                    logger.error(f"[批量历史消息] 数据完整性错误: {ie}")
                                    error_count += 1
                            except Exception as msg_err:
                                logger.error(f"[批量历史消息] 保存单条消息失败: {msg_err}")
                                error_count += 1
                        
                        logger.info(f"[批量历史消息] ✅ 批量保存完成: 成功{success_count}条, 重复跳过{duplicate_count}条, 失败{error_count}条")
                        
                        # 不转发到前端（历史消息不需要实时推送）
                        return
                        
                    except Exception as e:
                        logger.error(f"[批量历史消息] 处理失败: {e}", exc_info=True)
                
                # ✅ 特殊处理：会话同步（Agent同步会话列表时上报lastMessage）
                elif event == "chat_message" and direction == "req":
                    try:
                        logger.info(f"[会话同步] 收到来自 {client_id} 的会话同步消息")
                        
                        from module_admin.dao.seller_conversation_dao import SellerConversationDao
                        from datetime import datetime as dt_module
                        
                        chat_id = data.get('chat_id')
                        seller_id = data.get('seller_id')
                        seller_name = data.get('seller_name', '')
                        item_id = data.get('item_id')
                        item_title = data.get('item_title', '')
                        message_content = data.get('message_content', '')
                        message_time = data.get('message_time')
                        
                        # 构建conversation_id（优先使用chat_id）
                        conversation_id = chat_id or f"{seller_id}#{item_id}"
                        
                        # 保存或更新会话记录
                        async with self._db_session() as db:
                            await SellerConversationDao.create_or_update_conversation(
                                db=db,
                                client_id=client_id,
                                conversation_id=conversation_id,
                                chat_id=chat_id,
                                seller_id=seller_id,
                                seller_name=seller_name,
                                item_id=item_id,
                                item_title=item_title,
                                last_message=message_content,
                                last_message_time=message_time
                            )
                            await db.commit()
                            
                        logger.info(f"[会话同步] ✅ 已保存会话: chat_id={chat_id}, seller_id={seller_id}")
                        return
                        
                    except Exception as e:
                        logger.error(f"[会话同步] 处理失败: {e}", exc_info=True)
                
                # 特殊处理：聊天消息（Agent 推送的新消息）
                elif event == "chat_message" and direction == "res":
                    try:
                        # ✅ 处理发送失败的消息
                        if data.get('success') is False:
                            error_msg = data.get('error', '发送失败')
                            temp_id = data.get('temp_id')
                            mid = data.get('mid')
                            
                            logger.warning(f"[聊天消息] 收到发送失败通知: client_id={client_id}, error={error_msg}")
                            
                            # 推送给前端
                            from tcp_server.websocket_manager import websocket_manager
                            await websocket_manager.broadcast({
                                'type': 'message_event',  # ✅ 使用 message_event 类型
                                'client_id': client_id,
                                'event': 'failed',        # ✅ event 设为 failed
                                'data': {
                                    'temp_id': temp_id,
                                    'mid': mid,
                                    'error': error_msg,
                                    'seller_id': data.get('seller_id'),
                                    'item_id': data.get('item_id'),
                                    'chat_id': data.get('chat_id')
                                }
                            }, user_id=ClientManager.get_user_id_by_client_id(client_id))
                            return

                        logger.info(f"[聊天消息] 收到来自 {client_id} 的新消息")
                        
                        # ✅ 检查seller_id是否是加密的（包含==或/，且不是纯数字）
                        seller_id = data.get('seller_id', '')
                        item_id = data.get('item_id')
                        seller_name = data.get('seller_name', '')
                        seller_avatar = data.get('seller_avatar', '')
                        chat_id = data.get('chat_id') or None  # ✅ 获取chat_id，用于构建conversation_id（空字符串视为None）
                        message_content = data.get('message_content', '')
                        
                        # ✅ 保护逻辑：如果seller_name与message_content相同，说明这是系统消息，清空seller_name避免覆盖真实卖家昵称
                        # 例如："未拍摄「开箱视频」"、"[我已修改价格，等待你付款]"、"[未付款，你关闭了订单]"等
                        if seller_name and message_content and seller_name == message_content:
                            logger.info(f"[聊天消息] 🔔 识别为系统消息（seller_name==message_content），清空seller_name: {seller_name}")
                            seller_name = ""  # ✅ 清空seller_name，避免覆盖真实的卖家昵称
                        
                        # 🔍 调试日志：打印chat_id的值
                        logger.info(f"[TCP] 🔍 接收消息时的chat_id: chat_id={chat_id}, type={type(chat_id)}, seller_id={seller_id}")
                        
                        # 判断seller_id是否是加密的（包含特殊字符且不是纯数字）
                        is_encrypted = seller_id and (not seller_id.isdigit()) and ('==' in seller_id or '/' in seller_id or len(seller_id) > 20)
                        
                        # ✅ 如果seller_name为空，且有item_id，尝试通过匿名请求获取卖家信息
                        if not seller_name and item_id:
                            try:
                                from module_admin.service.goofish_anonymous_service import GoofishAnonymousService
                                
                                # 获取Redis客户端
                                redis = self._redis
                                if not redis:
                                    logger.warning(f"[聊天消息] Redis不可用，无法获取卖家信息")
                                else:
                                    # 通过匿名请求获取商品详情，提取卖家信息
                                    item_info = await GoofishAnonymousService.get_item_and_seller_info(redis, int(item_id))
                                    if item_info:
                                        # 如果seller_name为空，更新
                                        if not seller_name and item_info.get('seller_name'):
                                            seller_name = item_info['seller_name']
                                            logger.info(f"[聊天消息] ✅ 已获取卖家名称: {seller_name}")
                                        # 如果seller_avatar为空，更新
                                        if not seller_avatar and item_info.get('seller_avatar'):
                                            seller_avatar = item_info['seller_avatar']
                                        # 如果seller_id是加密的，也更新为纯数字
                                        if is_encrypted and item_info.get('seller_id'):
                                            old_seller_id = seller_id
                                            seller_id = str(item_info['seller_id'])
                                            logger.info(f"[聊天消息] ✅ 已解析加密seller_id: {old_seller_id[:20]}... -> {seller_id}")
                            except Exception as e:
                                logger.warning(f"[聊天消息] 获取卖家信息失败: {e}")
                        
                        # ✅ 如果seller_id是加密的，且有item_id，尝试通过匿名请求获取纯数字seller_id
                        if is_encrypted and item_id:
                            try:
                                from module_admin.service.goofish_anonymous_service import GoofishAnonymousService
                                
                                # 获取Redis客户端
                                redis = self._redis
                                if not redis:
                                    logger.warning(f"[聊天消息] Redis不可用，无法解析加密seller_id")
                                else:
                                    # 通过匿名请求获取商品详情，提取纯数字seller_id
                                    item_info = await GoofishAnonymousService.get_item_and_seller_info(redis, int(item_id))
                                    if item_info and item_info.get('seller_id'):
                                        # 更新为纯数字seller_id
                                        old_seller_id = seller_id
                                        seller_id = str(item_info['seller_id'])
                                        
                                        # 如果seller_name或seller_avatar为空，也更新
                                        if not seller_name and item_info.get('seller_name'):
                                            seller_name = item_info['seller_name']
                                        if not seller_avatar and item_info.get('seller_avatar'):
                                            seller_avatar = item_info['seller_avatar']
                                        
                                        logger.info(f"[聊天消息] ✅ 已解析加密seller_id: {old_seller_id[:20]}... -> {seller_id}")
                            except Exception as e:
                                logger.warning(f"[聊天消息] 解析加密seller_id失败: {e}，将使用原始seller_id")
                        
                        # 保存到数据库
                        from module_admin.service.seller_message_service import SellerMessageService
                        # ✅ 提前提取关键字段（在async with之前，避免作用域问题）
                        message_id = data.get('message_id')
                        # message_content 已在前面提取（line 366），不需要重复提取
                        message_time = data.get('message_time')
                        conversation_id = None  # 稍后设置
                        direction = data.get('direction', 'recv')
                        
                        # ✅ 使用async with确保数据库连接正确关闭
                        async with self._db_session() as db:
                            # ✅ 检查是否是临时消息ID（格式为 timestamp.TMP），如果是，尝试更新已存在的消息
                            is_temp_id = message_id and message_id.endswith('.TMP')
                            
                            # ✅ 如果是临时ID，尝试查找并更新已存在的消息（通过conversation_id、direction、message_content匹配）
                            if is_temp_id and data.get('direction') == 'send':
                                from module_admin.dao.seller_message_dao import SellerMessageDao
                                from sqlalchemy import select, and_
                                from module_admin.entity.do.seller_message_do import SellerMessage
                                
                                # 查找最近发送的相同内容的消息（可能是之前用临时ID保存的）
                                # ✅ 使用chat_id作为conversation_id（如果存在且有效），否则使用seller_id#item_id
                                # ⚠️ 重要：确保chat_id不为空字符串，且item_id存在时才使用fallback格式
                                if chat_id and chat_id.strip():
                                    conversation_id = chat_id
                                elif seller_id and item_id:
                                    conversation_id = f"{seller_id}#{item_id}"
                                else:
                                    logger.error(f"[TCP] ❌ 无法构建conversation_id: chat_id={chat_id}, seller_id={seller_id}, item_id={item_id}")
                                    conversation_id = seller_id or 'unknown'  # 临时值，应该避免这种情况
                                # message_content 已在前面提取（line 366），不需要重复提取
                                
                                # 查找最近5分钟内发送的相同内容的消息，且message_id是临时ID
                                from datetime import datetime, timedelta
                                five_min_ago = datetime.now() - timedelta(minutes=5)
                                
                                stmt = select(SellerMessage).where(
                                    and_(
                                        SellerMessage.conversation_id == conversation_id,
                                        SellerMessage.direction == 'send',
                                        SellerMessage.message_content == message_content,
                                        SellerMessage.message_time >= five_min_ago,
                                        SellerMessage.message_id.like('%.TMP')  # 临时ID格式
                                    )
                                ).order_by(SellerMessage.message_time.desc()).limit(1)
                                
                                result = await db.execute(stmt)
                                existing_msg = result.scalar_one_or_none()
                                
                                if existing_msg:
                                    # ✅ 注意：这里message_id还是临时ID，真实的PNM ID会在后续同步包中更新
                                    # 所以这里暂时不更新message_id，等收到真实ID时再更新
                                    # 但可以先更新message_extra（如果新消息有的话）
                                    # 实际上，如果找到了临时ID的消息，应该跳过创建新消息，等待后续同步包更新
                                    logger.debug(f"[聊天消息] 找到临时消息，跳过创建，等待后续同步包更新: {existing_msg.id}")
                                    # 不创建新消息，继续等待后续同步包中的真实消息
                            
                        # ✅ 构建message_extra JSON（包含回复消息ID、图片URL、语音URL等扩展信息）
                        message_extra_dict = {}
                        if data.get('message_extra'):
                            try:
                                # ✅ 使用文件顶部已导入的json模块，不需要重新导入
                                existing_extra = json.loads(data.get('message_extra'))
                                if isinstance(existing_extra, dict):
                                    message_extra_dict.update(existing_extra)
                            except Exception:
                                pass
                        
                        # ✅ 如果有回复消息ID，添加到message_extra中
                        reply_message_id = data.get('reply_message_id')
                        if reply_message_id:
                            message_extra_dict['reply_message_id'] = reply_message_id
                        
                        # ✅ 如果是图片消息，添加图片信息到message_extra
                        message_type = data.get('message_type', 'text')
                        if message_type == 'image':
                            image_url = data.get('image_url')  # ✅ 即使为None也获取
                            image_width = data.get('image_width')
                            image_height = data.get('image_height')
                            # ✅ 无论URL是否存在，都添加到message_extra
                            if image_url:  # 只有当URL存在时才添加
                                message_extra_dict['image_url'] = image_url
                            if image_width:
                                message_extra_dict['image_width'] = image_width
                            if image_height:
                                message_extra_dict['image_height'] = image_height
                            # ✅ 如果图片消息没有URL，记录错误并至少保存消息类型标记
                            if not image_url:
                                logger.error(f"[TCP] ❌ 图片消息缺少image_url: message_id={message_id}, message_type={message_type}, data_keys={list(data.keys())}, data={json.dumps(data, ensure_ascii=False)[:1000]}")
                                # ✅ 即使没有URL，也标记这是一个图片消息（避免message_extra完全为空）
                                message_extra_dict['message_type'] = 'image'
                                message_extra_dict['missing_url'] = True
                        
                        # ✅ 如果是语音消息，添加语音信息到message_extra
                        if message_type == 'voice':
                            voice_url = data.get('voice_url')  # ✅ 即使为None也获取
                            voice_duration = data.get('voice_duration')
                            voice_size_bytes = data.get('voice_size_bytes')
                            # ✅ 无论URL是否存在，都添加到message_extra
                            if voice_url:  # 只有当URL存在时才添加
                                message_extra_dict['voice_url'] = voice_url
                            if voice_duration is not None:
                                message_extra_dict['voice_duration'] = voice_duration
                            if voice_size_bytes is not None:
                                message_extra_dict['voice_size_bytes'] = voice_size_bytes
                            # ✅ 如果语音消息没有URL，记录错误并至少保存消息类型标记
                            if not voice_url:
                                logger.error(f"[TCP] ❌ 语音消息缺少voice_url: message_id={message_id}, message_type={message_type}, data_keys={list(data.keys())}, data={json.dumps(data, ensure_ascii=False)[:1000]}")
                                # ✅ 即使没有URL，也标记这是一个语音消息（避免message_extra完全为空）
                                message_extra_dict['message_type'] = 'voice'
                                message_extra_dict['missing_url'] = True
                        
                        # ✅ 如果是系统消息（textCard），添加系统消息数据到message_extra
                        if message_type == 'system':
                            system_data = data.get('system_data')
                            if system_data and isinstance(system_data, dict):
                                # ✅ 保存完整的系统消息数据
                                message_extra_dict['system_data'] = system_data
                                message_extra_dict['contentType'] = system_data.get('contentType', 6)
                                logger.info(f"[TCP] 🔔 保存系统消息: message_id={message_id}, title={system_data.get('title', '')[:50]}")
                        
                        message_extra_json = json.dumps(message_extra_dict, ensure_ascii=False) if message_extra_dict else None
                        
                        # ✅ 重要：如果message_type是image、voice或system但message_extra_json仍然为空，说明数据提取失败，至少保存一个标记
                        if not message_extra_json and message_type in ['image', 'voice', 'system']:
                            message_extra_dict = {'message_type': message_type, 'missing_data': True}
                            message_extra_json = json.dumps(message_extra_dict, ensure_ascii=False)
                            logger.error(f"[TCP] ❌ 严重：{message_type}消息的message_extra为空，已保存标记: message_id={message_id}")
                        
                        # ✅ 调试日志：如果message_extra_dict不为空，记录保存的内容
                        if message_extra_dict:
                            logger.debug(f"[TCP] ✅ message_extra已构建: message_id={message_id}, message_type={message_type}, extra_keys={list(message_extra_dict.keys())}")
                        else:
                            logger.debug(f"[TCP] ⚠️ message_extra为空: message_id={message_id}, message_type={message_type}, data_keys={list(data.keys())}")
                        
                        # 构建消息数据（使用可能已更新的seller_id）
                        # ✅ 使用chat_id作为conversation_id（如果存在且有效），否则查找数据库或使用临时格式
                        # ⚠️ 关键：多层fallback确保conversation_id永远不为NULL
                        conversation_id = None  # ✅ 初始化为None
                        
                        if chat_id and chat_id.strip():
                            conversation_id = chat_id
                            logger.debug(f"[TCP] ✅ 使用chat_id作为conversation_id: {conversation_id}")
                        elif seller_id:
                            # ✅ 优先从seller_conversation表查找真实chat_id（最可靠）
                            from sqlalchemy import select, and_, or_
                            from module_admin.entity.do.seller_conversation_do import SellerConversation
                            from module_admin.entity.do.seller_message_do import SellerMessage
                            
                            existing_chat_id = None
                            
                            # ✅ 步骤1：从seller_conversation表查找（会话同步时保存的真实chat_id）
                            if item_id:
                                conv_query = await db.execute(
                                    select(SellerConversation.chat_id)
                                    .where(
                                        SellerConversation.seller_id == seller_id,
                                        SellerConversation.item_id == item_id,
                                        SellerConversation.client_id == client_id,
                                        SellerConversation.chat_id.isnot(None),
                                        SellerConversation.chat_id != '',
                                        SellerConversation.del_flag == '0'
                                    )
                                    .order_by(SellerConversation.last_message_time.desc())
                                    .limit(1)
                                )
                                existing_chat_id = conv_query.scalar_one_or_none()
                                
                                if existing_chat_id:
                                    logger.info(f"[TCP] ✅ 从seller_conversation表查找到chat_id: {existing_chat_id}")
                            
                            # ✅ 步骤2：如果conversation表没有，从seller_message表查找
                            if not existing_chat_id and item_id:
                                msg_query = await db.execute(
                                    select(SellerMessage.conversation_id)
                                    .where(
                                        SellerMessage.seller_id == seller_id,
                                        SellerMessage.item_id == item_id,
                                        SellerMessage.client_id == client_id,
                                        SellerMessage.conversation_id.isnot(None),
                                        SellerMessage.conversation_id != '',
                                        SellerMessage.del_flag == '0'
                                        # ✅ 允许查找临时ID（移除#号过滤），后续会自动升级
                                    )
                                    .order_by(SellerMessage.create_time.desc())
                                    .limit(1)
                                )
                                existing_chat_id = msg_query.scalar_one_or_none()
                                
                                if existing_chat_id:
                                    logger.info(f"[TCP] ✅ 从seller_message表查找到conversation_id: {existing_chat_id}")
                            
                            # ✅ 步骤3：使用查找到的chat_id或生成临时格式
                            if existing_chat_id:
                                conversation_id = existing_chat_id
                            elif item_id:
                                conversation_id = f"{seller_id}#{item_id}"
                                logger.warning(f"[TCP] ⚠️ 未找到chat_id，使用临时格式: {conversation_id}")
                            else:
                                conversation_id = f"{seller_id}#unknown"
                                logger.warning(f"[TCP] ⚠️ 未找到chat_id且无item_id，使用临时格式: {conversation_id}")
                        
                        # ✅ 最终fallback：如果conversation_id仍然是None，使用chat_id或临时格式
                        # 确保conversation_id永远不为NULL
                        if not conversation_id:
                            if chat_id:
                                conversation_id = chat_id
                                logger.warning(f"[TCP] ⚠️ 使用chat_id作为fallback: {conversation_id}")
                            elif seller_id and item_id:
                                conversation_id = f"{seller_id}#{item_id}"
                                logger.warning(f"[TCP] ⚠️ 使用临时格式作为fallback: {conversation_id}")
                            else:
                                # ❌ 极端情况：所有条件都不满足，记录错误并跳过
                                logger.error(f"[TCP] ❌ 无法构建conversation_id，消息将被跳过: chat_id={chat_id}, seller_id={seller_id}, item_id={item_id}")
                                return
                        
                        # ✅ 如果有chat_id（真实会话ID），检查是否需要更新旧的临时conversation_id
                        if chat_id and seller_id and item_id:
                            fallback_conversation_id = f"{seller_id}#{item_id}"
                            if fallback_conversation_id != chat_id:
                                # ✅ 更新所有使用fallback格式的旧消息为真实的chat_id
                                from sqlalchemy import update, select, and_
                                from module_admin.entity.do.seller_message_do import SellerMessage
                                update_stmt = update(SellerMessage).where(
                                    and_(
                                        SellerMessage.conversation_id == fallback_conversation_id,
                                        SellerMessage.client_id == client_id,
                                        SellerMessage.seller_id == seller_id,
                                        SellerMessage.item_id == item_id
                                    )
                                ).values(conversation_id=chat_id)
                                
                                result = await db.execute(update_stmt)
                                updated_count = result.rowcount
                                if updated_count > 0:
                                    await db.commit()
                                    logger.info(f"✅ 已更新 {updated_count} 条旧消息的conversation_id: {fallback_conversation_id} -> {chat_id}")
                        
                        message_data = {
                            'conversation_id': conversation_id,
                            'chat_id': chat_id,  # ✅ 添加chat_id字段
                            'message_id': message_id,  # ✅ 使用真实的消息ID（格式为 "3802712875012.PNM"）
                            'client_id': client_id,
                            'client_name': data.get('client_name'),
                            'account_user_id': data.get('account_user_id'),
                            'seller_id': seller_id,  # 使用可能已更新的纯数字seller_id
                            'seller_name': seller_name,  # 使用可能已更新的seller_name
                            'item_id': item_id,
                            'item_title': data.get('item_title'),
                            'message_type': data.get('message_type', 'text'),  # 内容类型
                            'custom_content_type': data.get('custom_content_type'),  # 官方内容类型代码
                            'content_type': data.get('content_type', 'private'),  # 消息类型
                            # ✅ 重要：direction 必须从 Agent 传递，如果没有则默认为 'recv'
                            # Agent 发送消息时应该明确传递 direction='send'
                            'direction': data.get('direction', 'recv'),
                            'message_content': data.get('message_content'),
                            'message_extra': message_extra_json,  # ✅ 使用包含回复消息ID的JSON
                            'message_time': data.get('message_time')
                        }
                        
                        # ✅ 关键日志：打印保存前的message_extra值
                        logger.info(f"[TCP] 🔍 准备保存消息到数据库: message_id={message_id}, message_type={data.get('message_type')}, message_extra={message_extra_json}, message_extra长度={len(message_extra_json) if message_extra_json else 0}")
                        
                        # ✅ 检查消息是否已存在：存在则更新，不存在则创建
                        existing_message_by_id = None
                        if message_id and not is_temp_id:
                            # 检查数据库中是否已存在相同message_id的消息
                            from sqlalchemy import select
                            from module_admin.entity.do.seller_message_do import SellerMessage
                            
                            existing_check = await db.execute(
                                select(SellerMessage).where(SellerMessage.message_id == message_id).limit(1)
                            )
                            existing_message_by_id = existing_check.scalar_one_or_none()
                            
                            if existing_message_by_id:
                                logger.debug(f"[TCP] 📝 消息已存在，准备更新: message_id={message_id}, existing_id={existing_message_by_id.id}")
                        
                        # 处理消息：更新或创建
                        if existing_message_by_id:
                            # ✅ 消息已存在，更新它（补充完整信息）
                            update_data = {}
                            
                            # 如果新消息有message_extra但旧消息没有或不完整，更新它
                            if message_extra_json:
                                if not existing_message_by_id.message_extra or existing_message_by_id.message_extra.strip() == '':
                                    update_data['message_extra'] = message_extra_json
                                    logger.info(f"[TCP] 📋 补充message_extra: message_id={message_id}")
                                else:
                                    # 如果都有，比较内容，选择更完整的
                                    try:
                                        import json as json_module
                                        old_extra = json_module.loads(existing_message_by_id.message_extra)
                                        new_extra = json_module.loads(message_extra_json)
                                        # 如果新的有更多字段，使用新的
                                        if len(new_extra) > len(old_extra):
                                            update_data['message_extra'] = message_extra_json
                                            logger.info(f"[TCP] 📋 更新为更完整的message_extra: message_id={message_id}")
                                    except:
                                        pass
                            
                            # 更新custom_content_type（如果新消息有但旧消息没有）
                            if data.get('custom_content_type') and not existing_message_by_id.custom_content_type:
                                update_data['custom_content_type'] = data.get('custom_content_type')
                                logger.info(f"[TCP] 📋 补充custom_content_type: message_id={message_id}")
                            
                            # 更新content_type（如果新消息有但旧消息没有）
                            if data.get('content_type') and not existing_message_by_id.content_type:
                                update_data['content_type'] = data.get('content_type')
                                logger.info(f"[TCP] 📋 补充content_type: message_id={message_id}")
                            
                            # 如果有更新内容，执行更新
                            if update_data:
                                from module_admin.dao.seller_message_dao import SellerMessageDao
                                await SellerMessageDao.update_message(db, existing_message_by_id.id, update_data)
                                await db.commit()
                                logger.info(f"[TCP] ✅ 已更新消息: ID={existing_message_by_id.id}, message_id={message_id}, 更新字段={list(update_data.keys())}")
                                created_msg = existing_message_by_id  # 使用更新后的消息
                            else:
                                logger.debug(f"[TCP] ⏭️ 消息已是最新，无需更新: message_id={message_id}")
                                created_msg = existing_message_by_id  # 使用现有消息
                        else:
                            # ✅ 如果message_id是临时ID且是发送消息，尝试查找并更新已存在的消息
                            existing_msg = None
                            if is_temp_id and data.get('direction') == 'send':
                                # 查找最近发送的相同内容的消息（可能是之前用临时ID保存的）
                                # ✅ 使用chat_id作为conversation_id（如果存在），否则使用seller_id#item_id
                                conversation_id = chat_id if chat_id else f"{seller_id}#{item_id or ''}"
                                # message_content 已在前面提取（line 366），不需要重复提取
                                
                                # 查找最近5分钟内发送的相同内容的消息，且message_id是临时ID
                                from datetime import datetime, timedelta
                                from sqlalchemy import select, and_
                                from module_admin.entity.do.seller_message_do import SellerMessage
                                five_min_ago = datetime.now() - timedelta(minutes=5)
                                
                                stmt = select(SellerMessage).where(
                                    and_(
                                        SellerMessage.conversation_id == conversation_id,
                                        SellerMessage.direction == 'send',
                                        SellerMessage.message_content == message_content,
                                        SellerMessage.message_time >= five_min_ago,
                                        SellerMessage.message_id.like('%.TMP')  # 临时ID格式
                                    )
                                ).order_by(SellerMessage.message_time.desc()).limit(1)
                                
                                result = await db.execute(stmt)
                                existing_msg = result.scalar_one_or_none()
                                
                                if existing_msg:
                                    # ✅ 更新已存在消息的message_id和message_extra（如果后续收到真实ID或更完整的数据时会更新）
                                    # 注意：这里message_id还是临时ID，真实的PNM ID会在后续同步包中更新
                                    # ⚠️ 重要：如果新消息有message_extra但旧消息没有，必须更新！
                                    update_data = {}
                                    if message_extra_json and (not existing_msg.message_extra or existing_msg.message_extra.strip() == ''):
                                        # 如果旧消息没有message_extra，但新消息有，更新它
                                        update_data['message_extra'] = message_extra_json
                                        logger.info(f"[聊天消息] ✅ 更新临时消息的message_extra: 旧消息ID={existing_msg.id}, message_extra={message_extra_json[:100]}")
                                    # 如果新消息的消息类型不同，也更新
                                    if data.get('message_type') and data.get('message_type') != existing_msg.message_type:
                                        update_data['message_type'] = data.get('message_type')
                                        logger.info(f"[聊天消息] ✅ 更新临时消息的message_type: {existing_msg.message_type} -> {data.get('message_type')}")
                                    
                                    if update_data:
                                        from module_admin.dao.seller_message_dao import SellerMessageDao
                                        await SellerMessageDao.update_message(db, existing_msg.id, update_data)
                                        await db.commit()
                                        logger.info(f"[聊天消息] ✅ 已更新临时消息: ID={existing_msg.id}, 更新内容={update_data}")
                                    
                                    logger.debug(f"[聊天消息] 找到临时消息ID的消息，使用现有消息: ID={existing_msg.id}")
                                    created_msg = existing_msg
                                else:
                                    # 创建新消息
                                    created_msg = await SellerMessageService.create_message(
                                        db=db,
                                        data=message_data,
                                        create_by=0,
                                        dept_id=0
                                    )
                                    await db.commit()
                            else:
                                # 接收消息或真实ID的发送消息，直接创建
                                created_msg = await SellerMessageService.create_message(
                                    db=db,
                                    data=message_data,
                                    create_by=0,
                                    dept_id=0
                                )
                                await db.commit()
                            
                            # ✅ 验证保存后的message_extra是否正确
                            logger.info(f"[聊天消息] ✅ 已保存到数据库: ID={created_msg.id}, seller_id={seller_id}, message_id={message_id}, message_type={created_msg.message_type}, message_extra={created_msg.message_extra[:100] if created_msg.message_extra else 'NULL'}")
                            
                            # ✅ 自动创建或更新会话记录
                            try:
                                from module_admin.dao.seller_conversation_dao import SellerConversationDao
                                from datetime import datetime as dt_module
                                
                                # 构建conversation_id
                                final_conversation_id = conversation_id or f"{seller_id}#{item_id or ''}"
                                
                                # ✅ 从data或已获取的变量中获取字段
                                item_title_val = data.get('item_title') or None
                                item_image_val = data.get('item_image') or data.get('image_url') or None
                                item_price_val = data.get('item_price') or data.get('price') or None
                                message_content_val = data.get('message_content') or message_content
                                message_time_val = data.get('message_time') or message_time
                                
                                # ✅ 安全处理message_time
                                if isinstance(message_time_val, str):
                                    last_msg_time = dt_module.fromisoformat(message_time_val)
                                elif isinstance(message_time_val, dt_module):
                                    last_msg_time = message_time_val
                                else:
                                    last_msg_time = dt_module.now()
                                
                                await SellerConversationDao.create_or_update_conversation(
                                    db=db,
                                    client_id=client_id,
                                    conversation_id=final_conversation_id,
                                    chat_id=chat_id,
                                    seller_id=seller_id,
                                    seller_name=seller_name,
                                    seller_avatar=seller_avatar if seller_avatar else None,
                                    item_id=item_id,
                                    item_title=item_title_val,
                                    item_image=item_image_val,  # ✅ 修复：传递商品图片
                                    item_price=item_price_val,   # ✅ 修复：传递商品价格
                                    last_message=message_content_val,
                                    last_message_time=last_msg_time,
                                    last_message_direction=direction,
                                    unread_count=1 if direction == 'recv' else 0,  # 接收消息时增加未读数
                                )
                                logger.info(f"[聊天消息] ✅ 会话已更新: conversation_id={final_conversation_id}, seller_id={seller_id}, chat_id={chat_id}")
                            except Exception as conv_e:
                                logger.error(f"[聊天消息] ❌ 更新会话记录失败: {conv_e}", exc_info=True)
                            
                            # ✅ 如果收到真实的消息ID（格式为 "3802712875012.PNM"），尝试更新之前保存的临时ID消息
                            if message_id and message_id.endswith('.PNM') and not is_temp_id:
                                # 查找是否有相同内容的临时ID消息需要更新
                                # ✅ 使用chat_id作为conversation_id（如果存在），否则使用seller_id#item_id
                                conversation_id = chat_id if chat_id else f"{seller_id}#{item_id or ''}"
                                # message_content 已在前面提取（line 366），不需要重复提取
                                
                                # 查找最近5分钟内相同内容的临时ID消息
                                from datetime import datetime, timedelta
                                from sqlalchemy import select, and_
                                from module_admin.entity.do.seller_message_do import SellerMessage
                                five_min_ago = datetime.now() - timedelta(minutes=5)
                                
                                stmt = select(SellerMessage).where(
                                    and_(
                                        SellerMessage.conversation_id == conversation_id,
                                        SellerMessage.direction == data.get('direction'),
                                        SellerMessage.message_content == message_content,
                                        SellerMessage.message_time >= five_min_ago,
                                        SellerMessage.message_id.like('%.TMP')  # 临时ID格式
                                    )
                                ).order_by(SellerMessage.message_time.desc()).limit(1)
                                
                                result = await db.execute(stmt)
                                temp_msg = result.scalar_one_or_none()
                                
                                if temp_msg and temp_msg.id != created_msg.id:
                                    # ✅ 更新临时ID消息为真实ID和message_extra，并删除新创建的消息（避免重复）
                                    update_data = {'message_id': message_id}
                                    # ✅ 如果新消息有message_extra，也更新它（特别是图片和语音消息的URL）
                                    if message_extra_json:
                                        update_data['message_extra'] = message_extra_json
                                        logger.debug(f"[聊天消息] 同时更新message_extra: {message_extra_json[:100]}...")
                                    # ✅ 如果新消息的消息类型不同（比如从text变为image），也更新它
                                    if data.get('message_type') and data.get('message_type') != temp_msg.message_type:
                                        update_data['message_type'] = data.get('message_type')
                                        logger.debug(f"[聊天消息] 同时更新message_type: {temp_msg.message_type} -> {data.get('message_type')}")
                                    
                                    await SellerMessageDao.update_message(
                                        db, 
                                        temp_msg.id, 
                                        update_data
                                    )
                                    # 删除新创建的消息（因为已更新了临时消息）
                                    await SellerMessageDao.delete_message(db, created_msg.id)
                                    await db.commit()
                                    logger.info(f"[聊天消息] ✅ 已更新临时消息: {temp_msg.id} -> message_id={message_id}, message_type={data.get('message_type')}, has_extra={bool(message_extra_json)}")
                                    created_msg = temp_msg  # 使用更新后的消息
                            # ✅ 如果更新了临时消息，需要重新构建WebSocket消息数据（使用更新后的消息）
                            ws_message_data = message_data.copy()
                            ws_message_data['message_id'] = message_id  # 使用真实的消息ID
                            # ✅ 添加conversation_id和chat_id到WebSocket消息，确保前端可以正确关联会话
                            ws_message_data['conversation_id'] = conversation_id  # ✅ 关键：传递conversation_id
                            if chat_id:
                                ws_message_data['chat_id'] = chat_id
                            if seller_avatar:
                                ws_message_data['seller_avatar'] = seller_avatar
                            
                            # ✅ 如果是图片/语音/系统消息，从message_extra中提取数据并添加到WebSocket消息中
                            if message_extra_json:
                                try:
                                    message_extra_dict = json.loads(message_extra_json)
                                    # ✅ 图片消息：提取image_url等信息
                                    if data.get('message_type') == 'image':
                                        if message_extra_dict.get('image_url'):
                                            ws_message_data['image_url'] = message_extra_dict['image_url']
                                        if message_extra_dict.get('image_width'):
                                            ws_message_data['image_width'] = message_extra_dict['image_width']
                                        if message_extra_dict.get('image_height'):
                                            ws_message_data['image_height'] = message_extra_dict['image_height']
                                    # ✅ 语音消息：提取voice_url等信息
                                    elif data.get('message_type') == 'voice':
                                        if message_extra_dict.get('voice_url'):
                                            ws_message_data['voice_url'] = message_extra_dict['voice_url']
                                        if message_extra_dict.get('voice_duration'):
                                            ws_message_data['voice_duration'] = message_extra_dict['voice_duration']
                                        if message_extra_dict.get('voice_size_bytes'):
                                            ws_message_data['voice_size_bytes'] = message_extra_dict['voice_size_bytes']
                                    # ✅ 系统消息：提取system_data
                                    elif data.get('message_type') == 'system':
                                        if message_extra_dict.get('system_data'):
                                            ws_message_data['system_data'] = message_extra_dict['system_data']
                                            logger.debug(f"[聊天消息] ✅ 已添加system_data到WebSocket消息: message_id={message_id}")
                                except Exception as e:
                                    logger.warning(f"[聊天消息] ⚠️ 解析message_extra失败: {e}")
                            
                            # ✅ 推送到前端 WebSocket（包含conversation_id、chat_id、seller_avatar、system_data和图片/语音信息）
                            # 注意：如果上面更新了临时消息，ws_message_data已经重新构建；否则使用初始的message_data
                            if 'ws_message_data' not in locals():
                                ws_message_data = message_data.copy()
                                # ✅ 添加conversation_id和chat_id到WebSocket消息，确保前端可以正确关联会话
                                ws_message_data['conversation_id'] = conversation_id  # ✅ 关键：传递conversation_id
                                if chat_id:
                                    ws_message_data['chat_id'] = chat_id
                                if seller_avatar:
                                    ws_message_data['seller_avatar'] = seller_avatar
                                
                                # ✅ 如果是图片/语音/系统消息，从message_extra中提取数据并添加到WebSocket消息中
                                if message_extra_json:
                                    try:
                                        message_extra_dict = json.loads(message_extra_json)
                                        # ✅ 图片消息：提取image_url等信息
                                        if data.get('message_type') == 'image':
                                            if message_extra_dict.get('image_url'):
                                                ws_message_data['image_url'] = message_extra_dict['image_url']
                                            if message_extra_dict.get('image_width'):
                                                ws_message_data['image_width'] = message_extra_dict['image_width']
                                            if message_extra_dict.get('image_height'):
                                                ws_message_data['image_height'] = message_extra_dict['image_height']
                                        # ✅ 语音消息：提取voice_url等信息
                                        elif data.get('message_type') == 'voice':
                                            if message_extra_dict.get('voice_url'):
                                                ws_message_data['voice_url'] = message_extra_dict['voice_url']
                                            if message_extra_dict.get('voice_duration'):
                                                ws_message_data['voice_duration'] = message_extra_dict['voice_duration']
                                            if message_extra_dict.get('voice_size_bytes'):
                                                ws_message_data['voice_size_bytes'] = message_extra_dict['voice_size_bytes']
                                        # ✅ 系统消息：提取system_data
                                        elif data.get('message_type') == 'system':
                                            if message_extra_dict.get('system_data'):
                                                ws_message_data['system_data'] = message_extra_dict['system_data']
                                                logger.debug(f"[聊天消息] ✅ 已添加system_data到WebSocket消息: message_id={message_id}")
                                    except Exception as e:
                                        logger.warning(f"[聊天消息] ⚠️ 解析message_extra失败: {e}")
                            
                            # ✅ 只广播一次（避免重复广播）
                            # ✅ 获取user_id并过滤推送
                            user_id = ClientManager.get_user_id_by_client_id(client_id)
                            await ClientManager.broadcast_to_websocket({
                                'type': 'chat_message',
                                'client_id': client_id,
                                'data': ws_message_data
                            }, user_id=user_id)
                            logger.info(f"[聊天消息] ✅ 已广播到前端WebSocket: seller_id={seller_id}, direction={message_data.get('direction')}, message_id={message_id}, message_type={message_data.get('message_type')}")
                        
                        logger.info(f"[聊天消息] 已保存到数据库: ID={created_msg.id}, seller_id={seller_id}, message_id={message_id}")
                    except Exception as e:
                        logger.error(f"[聊天消息] 处理失败: {e}", exc_info=True)
                
                # ✅ 特殊处理：已读回执确认（Agent发送已读回执后闲鱼服务器的确认）
                elif event == "chat_read_confirmed" and direction == "res":
                    try:
                        logger.info(f"[已读回执确认] 收到来自 {client_id} 的已读回执确认")
                        
                        message_ids = data.get('message_ids', [])
                        success = data.get('success', False)
                        mid = data.get('mid')
                        error = data.get('error')
                        
                        if success and message_ids:
                            # ✅ 更新数据库中的read_status为已读（2）
                            from sqlalchemy import update, and_
                            from module_admin.entity.do.seller_message_do import SellerMessage
                            
                            async with self._db_session() as db:
                                update_stmt = update(SellerMessage).where(
                                    and_(
                                        SellerMessage.message_id.in_(message_ids),
                                        SellerMessage.client_id == client_id,
                                        SellerMessage.del_flag == '0'
                                    )
                                ).values(read_status=2)  # 2=已读
                                
                                result = await db.execute(update_stmt)
                                updated_count = result.rowcount
                                await db.commit()
                                
                                logger.info(f"[已读回执确认] ✅ 已更新 {updated_count} 条消息为已读状态: mid={mid}, message_ids={message_ids[:5]}...")
                                
                                # ✅ 广播已读回执确认到前端（告知用户已读回执发送成功）
                                user_id = ClientManager.get_user_id_by_client_id(client_id)
                                await ClientManager.broadcast_to_websocket({
                                    'type': 'chat_read_confirmed',
                                    'client_id': client_id,
                                    'data': {
                                        'message_ids': message_ids,
                                        'success': True,
                                        'updated_count': updated_count
                                    }
                                }, user_id=user_id)
                        else:
                            logger.warning(f"[已读回执确认] ⚠️ 已读回执发送失败: mid={mid}, error={error}, message_ids={message_ids[:5] if message_ids else []}")
                    except Exception as e:
                        logger.error(f"[已读回执确认] 处理失败: {e}", exc_info=True)
                
                # 特殊处理：取消订单响应需要更新订单状态
                if event == "cancel_order":
                    logger.info(f"📥 收到取消订单响应: success={message.get('success')}, data={data}, msg={message.get('msg')}")
                    try:
                        async with self._db_session() as db:
                            if message.get("success"):
                                # 取消成功
                                cancel_data = data or {}
                                order_id = cancel_data.get("order_id") or cancel_data.get("orderId") or ""
                                logger.info(f"✅ 取消订单成功响应: order_id={order_id}")
                                
                                if order_id:
                                    # ✅ 查找订单记录（通过biz_order_id）
                                    from module_admin.dao.search_task_dao import OrderRecordDao
                                    from sqlalchemy import select
                                    from module_admin.entity.do.order_record_do import OrderRecord
                                    
                                    order_record = await db.execute(
                                        select(OrderRecord).where(
                                            OrderRecord.biz_order_id == str(order_id),
                                            OrderRecord.del_flag == '0'
                                        )
                                    )
                                    order_record = order_record.scalar_one_or_none()
                                    
                                    if order_record:
                                        # ✅ 更新订单状态为cancelled
                                        await OrderRecordDao.update_record(db, order_record.id, {
                                            'order_status': 'cancelled',
                                            'error': None
                                        })
                                        
                                        await db.commit()
                                        logger.success(f"✅ 订单已取消并更新数据库: order_id={order_id}, item_id={order_record.item_id}, 已取消已下单标记")
                                        
                                        # ✅ 广播取消订单通知到前端WebSocket（按user_id过滤）
                                        cancel_user_id = order_record.create_by  # 从订单记录获取user_id
                                        if not cancel_user_id:
                                            logger.warning(f"⚠️ 订单取消通知：order_record.create_by为None, order_id={order_id}, order_record_id={order_record.id}")
                                        await ClientManager.broadcast_to_websocket({
                                            "type": "order_cancelled",
                                            "client_id": client_id,
                                            "data": {
                                                "order_id": order_id,
                                                "item_id": order_record.item_id,
                                                "order_record_id": order_record.id
                                            }
                                        }, user_id=cancel_user_id)
                                        if cancel_user_id:
                                            logger.info(f"✅ 订单取消通知已发送: order_id={order_id}, user_id={cancel_user_id}")
                                    else:
                                        logger.warning(f"⚠️ 取消订单成功，但未找到订单记录: order_id={order_id}")
                                else:
                                    logger.warning(f"⚠️ 取消订单成功响应中缺少order_id: cancel_data={cancel_data}")
                            else:
                                # 取消失败
                                error_code = data.get("code") or ""
                                error_msg = message.get("msg") or data.get("msg") or "取消失败"
                                logger.error(f"❌ 取消订单失败响应: code={error_code}, msg={error_msg}, data={data}")
                                
                                # ✅ 判断是否是订单不存在（被手机app取消）
                                # 注意：这里可能需要根据实际API返回的错误码判断
                                # 常见的错误码可能是：订单不存在、订单已取消等
                                if "不存在" in error_msg or "not found" in error_msg.lower() or "已取消" in error_msg:
                                    # 订单不存在或已被取消，需要提示前端是否删除订单记录
                                    logger.warning(f"订单可能已被手机app取消: {error_msg}")
                                    # 这里可以通过WebSocket向前端推送消息，提示用户是否删除订单
                                    # 或者前端可以通过取消订单API返回的code判断
                    except Exception as e:
                        logger.error(f"处理取消订单响应失败: {e}", exc_info=True)
                
                # 特殊处理：订单响应需要创建订单记录
                if event == "order":
                    try:
                        async with self._db_session() as db:
                            if message.get("success"):
                                # 订单成功，创建订单记录
                                order_data = data or {}
                                item_id = order_data.get("item_id")
                                order_result_data = order_data.get("order_data", {})
                                order_id = order_data.get("order_id")
                                
                                if item_id:
                                    # ✅ 获取商品信息（从SearchRecord获取，SearchRecord作为商品共享池）
                                    from module_admin.entity.do.search_record_do import SearchRecord
                                    from sqlalchemy import select
                                    
                                    search_item = None
                                    # ✅ 直接从SearchRecord获取商品信息
                                    query = select(SearchRecord).where(
                                        SearchRecord.item_id == str(item_id),
                                        SearchRecord.del_flag == '0'
                                    )
                                    result = await db.execute(query)
                                    search_record = result.scalar_one_or_none()
                                    
                                    if search_record:
                                        search_item = {
                                            "item_id": search_record.item_id,
                                            "title": search_record.title,
                                            "price": str(search_record.price) if search_record.price else None,
                                            "seller_id": search_record.seller_id,
                                            "user_nick_name": search_record.seller_nick,
                                            "area": search_record.location,
                                            "publish_time": search_record.publish_time,
                                            "position": search_record.position,
                                            "pic_url": search_record.image_url,
                                            "keyword": search_record.search_keyword
                                        }
                                    
                                    # 如果还是没有商品信息，尝试从order_result_data中提取
                                    if not search_item:
                                        logger.warning(f"无法从数据库获取商品信息，尝试从订单数据中提取: {item_id}")
                                        search_item = {
                                            "item_id": item_id,
                                            "title": order_result_data.get("title") or "",
                                            "price": order_result_data.get("price") or "",
                                        }
                                    
                                    # 获取客户端信息
                                    from module_admin.dao.agent_client_dao import AgentClientDao
                                    client_info = await AgentClientDao.get_client_by_client_id(db, client_id)
                                    
                                    # ✅ 获取user_id（用于设置订单记录的create_by字段）
                                    order_user_id = ClientManager.get_user_id_by_client_id(client_id)
                                    if not order_user_id and client_info:
                                        # 如果缓存中没有，从数据库获取
                                        order_user_id = client_info.assigned_user_id or client_info.create_by
                                    
                                    if client_info:
                                        # 获取账号信息
                                        # 阿里系用户ID：优先从cookie的unb字段获取，其次使用account字段
                                        account_user_id = ""
                                        
                                        try:
                                            # 1. 优先从cookie中获取unb字段（阿里系用户ID）
                                            if client_info.cookies:
                                                try:
                                                    cookies_dict = json.loads(client_info.cookies) if isinstance(client_info.cookies, str) else client_info.cookies
                                                    if isinstance(cookies_dict, dict):
                                                        # unb是阿里系的用户ID字段
                                                        if "unb" in cookies_dict:
                                                            account_user_id = str(cookies_dict["unb"])
                                                except (json.JSONDecodeError, TypeError, AttributeError) as e:
                                                    logger.debug(f"解析cookie失败: {e}")
                                            
                                            # 2. 如果cookie中没有unb，使用account字段（account也是用户ID）
                                            if not account_user_id:
                                                account_user_id = getattr(client_info, 'account', None) or ""
                                            
                                            user_name = getattr(client_info, 'account_nickname', None) or getattr(client_info, 'account', None) or getattr(client_info, 'client_name', None) or ""
                                        except AttributeError as e:
                                            logger.error(f"访问client_info属性失败: {e}, client_info类型={type(client_info)}, 可用属性={dir(client_info) if client_info else 'None'}", exc_info=True)
                                            # 使用默认值
                                            account_user_id = ""
                                            user_name = ""
                                        
                                        # 构造订单结果数据（从API响应中提取，兼容多种命名方式）
                                        # 注意：API响应可能使用驼峰命名（orderId, payUrl）或蛇形命名（order_id, pay_url）
                                        order_result = {
                                            "bizOrderId": (order_result_data.get("orderId") or 
                                                          order_result_data.get("bizOrderId") or 
                                                          order_result_data.get("order_id") or 
                                                          order_result_data.get("biz_order_id") or 
                                                          order_id or ""),
                                            "bizOrderIdStr": (order_result_data.get("orderIdStr") or 
                                                            order_result_data.get("bizOrderIdStr") or 
                                                            order_result_data.get("order_id_str") or 
                                                            order_result_data.get("biz_order_id_str") or ""),
                                            "payOrderId": (order_result_data.get("payOrderId") or 
                                                         order_result_data.get("pay_order_id") or ""),
                                            "payUrl": (order_result_data.get("payUrl") or 
                                                      order_result_data.get("pay_url") or ""),
                                            "serviceOrder": (order_result_data.get("serviceOrder") or 
                                                           order_result_data.get("service_order") or ""),
                                            "usePayUrl": (order_result_data.get("usePayUrl") or 
                                                         order_result_data.get("use_pay_url") or "")
                                        }
                                        
                                        # ✅ 1. 查找是否有pending状态的订单记录（正在下单）
                                        from module_admin.dao.search_task_dao import OrderRecordDao
                                        pending_record = await OrderRecordDao.get_pending_record_by_item_id(db, str(item_id))
                                        
                                        if pending_record:
                                            # ✅ 2. 更新pending订单记录为ordered状态（订单已创建成功）
                                            from module_admin.service.search_task_service import OrderRecordService
                                            # ✅ 使用文件顶部已导入的json模块，不需要重新导入
                                            
                                            update_data = {
                                                'order_status': 'ordered',
                                                'biz_order_id': order_result.get("bizOrderId") or "",
                                                'biz_order_id_str': order_result.get("bizOrderIdStr") or "",
                                                'pay_order_id': order_result.get("payOrderId") or "",
                                                'pay_url': order_result.get("payUrl") or "",
                                                'service_order': order_result.get("serviceOrder") or "",
                                                'use_pay_url': order_result.get("usePayUrl") or "",
                                                'error': None  # 清除错误信息
                                            }
                                            
                                            await OrderRecordDao.update_record(db, pending_record.id, update_data)
                                            await db.commit()
                                            logger.success(f"✅ 已更新pending订单记录为ordered状态: item_id={item_id}, order_id={pending_record.id}, biz_order_id={order_result.get('bizOrderId')}")
                                            
                                            await db.commit()
                                            
                                            # ✅ 广播下单成功通知到前端WebSocket（按user_id过滤）
                                            order_user_id = pending_record.create_by  # 从订单记录获取user_id
                                            if not order_user_id:
                                                logger.warning(f"⚠️ 订单成功通知：pending_record.create_by为None, item_id={item_id}, order_record_id={pending_record.id}")
                                            await ClientManager.broadcast_to_websocket({
                                                "type": "order_success",
                                                "client_id": client_id,
                                                "data": {
                                                    "order_id": order_result.get("bizOrderId"),
                                                    "item_id": item_id,
                                                    "order_record_id": pending_record.id,
                                                    "pay_url": order_result.get("payUrl")
                                                }
                                            }, user_id=order_user_id)
                                            logger.info(f"✅ 下单成功通知已广播到前端: order_id={order_result.get('bizOrderId')}, user_id={order_user_id}")
                                        else:
                                            # ✅ 2. 如果没有pending记录，创建新订单记录（可能是手动下单或其他情况）
                                            from module_admin.service.search_task_service import OrderRecordService
                                            
                                            try:
                                                created_record = await OrderRecordService.create_order_record(
                                                    db=db,
                                                    search_item=search_item,
                                                    account_user_id=account_user_id,
                                                    user_name=user_name,
                                                    order_result=order_result,
                                                    client_id=client_id,
                                                    create_by=order_user_id or 1,  # ✅ 使用从client_id获取的user_id
                                                    dept_id=1,    # TODO: 从客户端信息中获取
                                                    is_auto_order=False  # ✅ 没有pending记录，可能是手动下单
                                                )
                                                await db.commit()
                                                logger.success(f"✅ 订单记录已创建: item_id={item_id}, order_id={order_id}")
                                                
                                                # ✅ 广播下单成功通知到前端WebSocket（按user_id过滤）
                                                order_user_id = created_record.create_by  # 从订单记录获取user_id
                                                if not order_user_id:
                                                    logger.warning(f"⚠️ 订单成功通知：created_record.create_by为None, item_id={item_id}, order_record_id={created_record.id}")
                                                await ClientManager.broadcast_to_websocket({
                                                    "type": "order_success",
                                                    "client_id": client_id,
                                                    "data": {
                                                        "order_id": order_result.get("bizOrderId"),
                                                        "item_id": item_id,
                                                        "order_record_id": created_record.id,
                                                        "pay_url": order_result.get("payUrl")
                                                    }
                                                }, user_id=order_user_id)
                                                if order_user_id:
                                                    logger.info(f"✅ 下单成功通知已广播到前端: order_id={order_result.get('bizOrderId')}, user_id={order_user_id}")
                                            except Exception as e:
                                                await db.rollback()
                                                logger.error(f"创建订单记录失败: {e}", exc_info=True)
                                    else:
                                        logger.warning(f"无法获取客户端信息: client_id={client_id}")
                                else:
                                    logger.warning(f"订单响应缺少item_id: {data}")
                            else:
                                # 订单失败，记录日志
                                error_msg = message.get("msg", "未知错误")
                                logger.error(f"订单创建失败: {error_msg}")
                    except Exception as e:
                        logger.error(f"处理订单响应失败: {e}", exc_info=True)
                
                # ========== 消息状态机：处理 message_event ==========
                elif event == "message_event" and direction == "res":
                    try:
                        logger.info(f"[消息事件] 收到来自 {client_id} 的消息事件")
                        
                        event_type = data.get('event')  # message_created, message_confirmed, message_failed, message_recalled
                        event_data = data.get('data', {})
                        
                        logger.info(f"[消息事件] event_type={event_type}, temp_id={event_data.get('temp_id')}, mid={event_data.get('mid')}, message_id={event_data.get('message_id')}")
                        
                        # ✅ 根据事件类型处理
                        if event_type == "message_created":
                            # ========== message_created：保存到数据库（status=PENDING）==========
                            from module_admin.service.seller_message_service import SellerMessageService
                            from module_admin.dao.seller_conversation_dao import SellerConversationDao
                            from datetime import datetime as dt_module
                            
                            async with self._db_session() as db:
                                try:
                                    # ✅ 尝试解析真实的chat_id和conversation_id（避免重复会话）
                                    # 如果conversation_id是临时的（包含#），尝试从数据库查找已存在的chat_id
                                    resolved_conversation_id = event_data.get('conversation_id')
                                    resolved_chat_id = event_data.get('chat_id')
                                    
                                    if (not resolved_chat_id or '#' in str(resolved_conversation_id)) and event_data.get('seller_id') and event_data.get('item_id'):
                                        from module_admin.entity.do.seller_conversation_do import SellerConversation
                                        from sqlalchemy import select
                                        
                                        # 查找已存在的chat_id
                                        query = select(SellerConversation.chat_id).where(
                                            SellerConversation.seller_id == event_data.get('seller_id'),
                                            SellerConversation.item_id == event_data.get('item_id'),
                                            SellerConversation.client_id == event_data.get('client_id'),
                                            SellerConversation.chat_id.isnot(None),
                                            SellerConversation.chat_id != '',
                                            SellerConversation.del_flag == '0'
                                        ).limit(1)
                                        result = await db.execute(query)
                                        existing_chat_id = result.scalar_one_or_none()
                                        
                                        if existing_chat_id:
                                            resolved_chat_id = existing_chat_id
                                            resolved_conversation_id = existing_chat_id
                                            event_data['conversation_id'] = resolved_conversation_id  # 更新event_data
                                            event_data['chat_id'] = resolved_chat_id
                                            logger.info(f"[消息事件] ✅ 已修正conversation_id: {event_data.get('conversation_id')} -> {resolved_conversation_id}")

                                    # 1. 保存消息（status=PENDING）
                                    message_dict = {
                                        "client_id": event_data.get('client_id'),
                                        "seller_id": event_data.get('seller_id'),
                                        "item_id": event_data.get('item_id'),
                                        "chat_id": resolved_chat_id,  # ✅ 使用解析后的ID
                                        "conversation_id": resolved_conversation_id,  # ✅ 使用解析后的ID
                                        "message_id": event_data.get('message_id'),  # 临时ID（如 "local-xxx.TMP"）
                                        "temp_id": event_data.get('temp_id'),  # ← 关键！
                                        "mid": event_data.get('mid'),  # ← 关键！
                                        "message_status": 1,  # PENDING
                                        "direction": "send",
                                        "message_type": event_data.get('message_type', 'text'),
                                        "message_content": event_data.get('message_content'),
                                        "message_time": event_data.get('message_time'),
                                    }
                                    
                                    created_msg = await SellerMessageService.create_message(db, message_dict)
                                    
                                    logger.info(f"[消息事件] ✅ message_created 已保存: id={created_msg.id}, temp_id={event_data.get('temp_id')}, message_status=PENDING")
                                    
                                    # 2. ✅ 创建或更新会话记录（防止出现"暂无消息"的会话）
                                    try:
                                        conversation_id = event_data.get('conversation_id')
                                        chat_id = event_data.get('chat_id')
                                        seller_id = event_data.get('seller_id')
                                        item_id = event_data.get('item_id')
                                        message_content = event_data.get('message_content', '')
                                        message_time_str = event_data.get('message_time')
                                        
                                        # 解析消息时间
                                        try:
                                            last_msg_time = dt_module.fromisoformat(message_time_str) if message_time_str else dt_module.now()
                                        except Exception:
                                            last_msg_time = dt_module.now()
                                        
                                        conv = await SellerConversationDao.create_or_update_conversation(
                                            db=db,
                                            client_id=event_data.get('client_id'),
                                            conversation_id=conversation_id,
                                            chat_id=chat_id,
                                            seller_id=seller_id,
                                            seller_name=None,  # message_created时还没有seller_name
                                            seller_avatar=None,
                                            item_id=item_id,
                                            item_title=None,
                                            item_image=None,
                                            item_price=None,
                                            last_message=message_content[:100],  # 截取前100字符
                                            last_message_time=last_msg_time,
                                            last_message_direction='send',
                                            unread_count=0,  # 发送消息不增加未读数
                                        )
                                        
                                        # ✅ 确保消息的conversation_id与会话保持一致（DAO层可能有更强的去重逻辑）
                                        if conv and conv.conversation_id != created_msg.conversation_id:
                                            logger.info(f"[消息事件] 🔄 同步消息conversation_id: {created_msg.conversation_id} -> {conv.conversation_id}")
                                            created_msg.conversation_id = conv.conversation_id
                                            if conv.chat_id:
                                                created_msg.chat_id = conv.chat_id
                                            
                                            # 更新event_data，确保广播给前端的是最终一致的ID
                                            event_data['conversation_id'] = conv.conversation_id
                                            if conv.chat_id:
                                                event_data['chat_id'] = conv.chat_id
                                        
                                        logger.info(f"[消息事件] ✅ 会话记录已创建/更新: conversation_id={conv.conversation_id if conv else conversation_id}")
                                    except Exception as conv_e:
                                        logger.error(f"[消息事件] ⚠️ 更新会话记录失败: {conv_e}", exc_info=True)
                                        # 不阻塞消息保存，继续执行
                                    
                                    await db.commit()
                                    
                                except Exception as e:
                                    await db.rollback()
                                    logger.error(f"[消息事件] 保存 message_created 失败: {e}", exc_info=True)
                        
                        elif event_type == "message_confirmed":
                            # ========== message_confirmed：更新数据库（status=SENT，更新message_id）==========
                            from module_admin.dao.seller_message_dao import SellerMessageDao
                            from module_admin.dao.seller_conversation_dao import SellerConversationDao
                            from sqlalchemy import select, and_
                            from module_admin.entity.do.seller_message_do import SellerMessage
                            from datetime import datetime as dt_module
                            
                            async with self._db_session() as db:
                                try:
                                    # 查找临时消息（通过 temp_id 或 mid）
                                    stmt = select(SellerMessage).where(
                                        and_(
                                            SellerMessage.client_id == event_data.get('client_id'),
                                            (SellerMessage.temp_id == event_data.get('temp_id')) | (SellerMessage.mid == event_data.get('mid'))
                                        )
                                    ).order_by(SellerMessage.id.desc()).limit(1)
                                    
                                    result = await db.execute(stmt)
                                    existing_msg = result.scalar_one_or_none()
                                    
                                    if existing_msg:
                                        # ✅ 尝试解析真实的chat_id和conversation_id
                                        resolved_conversation_id = event_data.get('conversation_id')
                                        resolved_chat_id = event_data.get('chat_id')
                                        
                                        if (not resolved_chat_id or '#' in str(resolved_conversation_id)) and event_data.get('seller_id') and event_data.get('item_id'):
                                            from module_admin.entity.do.seller_conversation_do import SellerConversation
                                            
                                            query = select(SellerConversation.chat_id).where(
                                                SellerConversation.seller_id == event_data.get('seller_id'),
                                                SellerConversation.item_id == event_data.get('item_id'),
                                                SellerConversation.client_id == event_data.get('client_id'),
                                                SellerConversation.chat_id.isnot(None),
                                                SellerConversation.chat_id != '',
                                                SellerConversation.del_flag == '0'
                                            ).limit(1)
                                            result = await db.execute(query)
                                            existing_chat_id = result.scalar_one_or_none()
                                            
                                            if existing_chat_id:
                                                resolved_chat_id = existing_chat_id
                                                resolved_conversation_id = existing_chat_id
                                                event_data['conversation_id'] = resolved_conversation_id
                                                event_data['chat_id'] = resolved_chat_id
                                        
                                        # 更新为真实消息ID和status=SENT
                                        real_message_id = event_data.get('message_id')
                                        existing_msg.message_id = real_message_id
                                        existing_msg.message_status = 2  # SENT
                                        existing_msg.mid = event_data.get('mid')  # 确保mid存在
                                        
                                        # ✅ 如果解析出了更好的conversation_id，也更新消息的conversation_id
                                        if resolved_conversation_id and existing_msg.conversation_id != resolved_conversation_id:
                                            existing_msg.conversation_id = resolved_conversation_id
                                            if resolved_chat_id:
                                                existing_msg.chat_id = resolved_chat_id
                                            logger.info(f"[消息事件] ✅ 同时修复消息conversation_id: {existing_msg.conversation_id} -> {resolved_conversation_id}")
                                        
                                        logger.info(f"[消息事件] ✅ message_confirmed 已更新: id={existing_msg.id}, temp_id={event_data.get('temp_id')}, real_message_id={real_message_id}, message_status=SENT")
                                        
                                        # ✅ 同时更新会话记录的 last_message（使用真实message_id）
                                        try:
                                            conversation_id = resolved_conversation_id  # ✅ 使用解析后的ID
                                            chat_id = resolved_chat_id  # ✅ 使用解析后的ID
                                            message_content = event_data.get('message_content', '')
                                            message_time_str = event_data.get('message_time')
                                            
                                            # 解析消息时间
                                            try:
                                                last_msg_time = dt_module.fromisoformat(message_time_str) if message_time_str else dt_module.now()
                                            except Exception:
                                                last_msg_time = dt_module.now()
                                            
                                            conv = await SellerConversationDao.create_or_update_conversation(
                                                db=db,
                                                client_id=event_data.get('client_id'),
                                                conversation_id=conversation_id,
                                                chat_id=chat_id,
                                                seller_id=event_data.get('seller_id'),
                                                seller_name=None,  # 保持原有值（不更新）
                                                seller_avatar=None,
                                                item_id=event_data.get('item_id'),
                                                item_title=None,
                                                item_image=None,
                                                item_price=None,
                                                last_message=message_content[:100],  # 截取前100字符
                                                last_message_time=last_msg_time,
                                                last_message_direction='send',
                                                unread_count=0,  # 发送消息不增加未读数
                                            )
                                            
                                            # ✅ 确保消息的conversation_id与会话保持一致
                                            if conv and conv.conversation_id != existing_msg.conversation_id:
                                                logger.info(f"[消息事件] 🔄 同步confirmed消息conversation_id: {existing_msg.conversation_id} -> {conv.conversation_id}")
                                                existing_msg.conversation_id = conv.conversation_id
                                                if conv.chat_id:
                                                    existing_msg.chat_id = conv.chat_id
                                                
                                                # 更新event_data
                                                event_data['conversation_id'] = conv.conversation_id
                                                if conv.chat_id:
                                                    event_data['chat_id'] = conv.chat_id
                                                
                                                # 更新广播用的conversation_id变量
                                                conversation_id = conv.conversation_id
                                            
                                            logger.info(f"[消息事件] ✅ 会话记录已更新（message_confirmed）: conversation_id={conv.conversation_id if conv else conversation_id}, real_message_id={real_message_id}")
                                        except Exception as conv_e:
                                            logger.error(f"[消息事件] ⚠️ 更新会话记录失败（message_confirmed）: {conv_e}", exc_info=True)
                                        
                                        await db.commit()
                                        
                                        # ✅ 广播 message_confirmed 事件给前端
                                        from tcp_server.websocket_manager import websocket_manager
                                        target_user_id = ClientManager.get_user_id_by_client_id(event_data.get('client_id'))
                                        
                                        # 🚨 调试日志：记录广播目标
                                        if target_user_id is None:
                                            # 如果未分配用户，尝试默认发送给管理员(ID=1)以确保消息能送达
                                            logger.warning(f"[消息事件] ⚠️ Client {event_data.get('client_id')} 未分配用户，广播目标修正为管理员(user_id=1)")
                                            target_user_id = 1
                                        
                                        logger.info(f"[消息事件] 🚀 广播 confirmed: client_id={event_data.get('client_id')}, target_user_id={target_user_id}, mid={event_data.get('mid')}")
                                        
                                        await websocket_manager.broadcast({
                                            'type': 'message_event',
                                            'client_id': event_data.get('client_id'),
                                            'event': 'confirmed',
                                            'data': {
                                                'temp_id': event_data.get('temp_id'),
                                                'mid': event_data.get('mid'),
                                                'message_id': real_message_id,
                                                'conversation_id': conversation_id,
                                                'chat_id': event_data.get('chat_id'),
                                                'seller_id': event_data.get('seller_id'),
                                                'item_id': event_data.get('item_id')
                                            }
                                        }, user_id=target_user_id)
                                    else:
                                        logger.warning(f"[消息事件] ⚠️ message_confirmed 未找到对应消息: temp_id={event_data.get('temp_id')}, mid={event_data.get('mid')}")
                                    
                                except Exception as e:
                                    await db.rollback()
                                    logger.error(f"[消息事件] 更新 message_confirmed 失败: {e}", exc_info=True)
                        
                        elif event_type == "message_failed":
                            # ========== message_failed：更新数据库（status=FAILED）==========
                            from module_admin.dao.seller_message_dao import SellerMessageDao
                            from sqlalchemy import select, and_
                            from module_admin.entity.do.seller_message_do import SellerMessage
                            
                            async with self._db_session() as db:
                                try:
                                    # 查找临时消息（通过 temp_id 或 mid）
                                    stmt = select(SellerMessage).where(
                                        and_(
                                            SellerMessage.client_id == event_data.get('client_id'),
                                            (SellerMessage.temp_id == event_data.get('temp_id')) | (SellerMessage.mid == event_data.get('mid'))
                                        )
                                    ).order_by(SellerMessage.id.desc()).limit(1)
                                    
                                    result = await db.execute(stmt)
                                    existing_msg = result.scalar_one_or_none()
                                    
                                    if existing_msg:
                                        # 更新为status=FAILED
                                        existing_msg.message_status = 6  # FAILED
                                        # 可以将错误信息保存到message_extra
                                        error_msg = event_data.get('error', '未知错误')
                                        if existing_msg.message_extra:
                                            try:
                                                extra = json.loads(existing_msg.message_extra)
                                                extra['error'] = error_msg
                                                existing_msg.message_extra = json.dumps(extra, ensure_ascii=False)
                                            except Exception:
                                                existing_msg.message_extra = json.dumps({'error': error_msg}, ensure_ascii=False)
                                        else:
                                            existing_msg.message_extra = json.dumps({'error': error_msg}, ensure_ascii=False)
                                        
                                        await db.commit()
                                        
                                        logger.info(f"[消息事件] ✅ message_failed 已更新: id={existing_msg.id}, temp_id={event_data.get('temp_id')}, error={error_msg}")
                                        
                                        # ✅ 广播 message_failed 事件给前端
                                        from tcp_server.websocket_manager import websocket_manager
                                        target_user_id = ClientManager.get_user_id_by_client_id(event_data.get('client_id'))
                                        
                                        # 🚨 调试日志：记录广播目标
                                        if target_user_id is None:
                                            logger.warning(f"[消息事件] ⚠️ Client {event_data.get('client_id')} 未分配用户，广播目标修正为管理员(user_id=1)")
                                            target_user_id = 1
                                            
                                        logger.info(f"[消息事件] 🚀 广播 failed: client_id={event_data.get('client_id')}, target_user_id={target_user_id}, error={error_msg}")

                                        await websocket_manager.broadcast({
                                            'type': 'message_event',
                                            'client_id': event_data.get('client_id'),
                                            'event': 'failed',
                                            'data': {
                                                'temp_id': event_data.get('temp_id'),
                                                'mid': event_data.get('mid'),
                                                'error': error_msg,
                                                'conversation_id': existing_msg.conversation_id,
                                                'chat_id': existing_msg.chat_id,
                                                'seller_id': existing_msg.seller_id,
                                                'item_id': existing_msg.item_id
                                            }
                                        }, user_id=target_user_id)
                                    else:
                                        logger.warning(f"[消息事件] ⚠️ message_failed 未找到对应消息: temp_id={event_data.get('temp_id')}, mid={event_data.get('mid')}")
                                    
                                except Exception as e:
                                    await db.rollback()
                                    logger.error(f"[消息事件] 更新 message_failed 失败: {e}", exc_info=True)
                        
                        # ========== 转发到前端 WebSocket ==========
                        # ✅ 获取user_id并过滤推送
                        event_client_id = event_data.get('client_id')
                        user_id = ClientManager.get_user_id_by_client_id(event_client_id) if event_client_id else None
                        
                        if user_id is None:
                            logger.warning(f"[消息事件] ⚠️ Client {event_client_id} 未分配用户，广播目标修正为管理员(user_id=1)")
                            user_id = 1
                            
                        logger.info(f"[消息事件] 🚀 广播 message_created: client_id={event_client_id}, target_user_id={user_id}, mid={event_data.get('mid')}")
                        
                        await ClientManager.broadcast_to_websocket({
                            "type": "message_event",
                            "client_id": event_client_id,
                            "event": event_type,
                            "data": event_data
                        }, user_id=user_id)
                        
                        logger.info(f"[消息事件] ✅ 已转发到前端: event_type={event_type}, client_id={event_data.get('client_id')}")
                        
                    except Exception as e:
                        logger.error(f"[消息事件] 处理失败: {e}", exc_info=True)
                
                # 转发到WebSocket
                # ✅ 跳过chat_message和message_event事件（已在上面的处理逻辑中广播过，避免重复）
                if event not in ("chat_message", "message_event"):
                    # ✅ get_address事件需要推送给发送命令的用户和管理员
                    if event == "get_address":
                        user_id = ClientManager.get_user_id_by_client_id(client_id) if client_id else None
                        
                        # 推送给发送命令的用户
                        if user_id:
                            await ClientManager.broadcast_to_websocket({
                                "type": "client_response",
                                "client_id": client_id or "",
                                "event": event,
                                "message": message
                            }, user_id=user_id)
                            logger.debug(f"[WebSocket] 📍 get_address响应已推送给user_id={user_id}")
                        
                        # 额外推送给管理员（user_id=1）
                        if user_id != 1:  # 避免重复推送
                            await ClientManager.broadcast_to_websocket({
                                "type": "client_response",
                                "client_id": client_id or "",
                                "event": event,
                                "message": message
                            }, user_id=1)
                            logger.debug(f"[WebSocket] 📍 get_address响应已推送给管理员(user_id=1)")
                    else:
                        # ✅ 其他事件：获取user_id并过滤推送（业务消息需要按用户隔离）
                        # 优先级：1. 从message.data中获取（搜索响应可能包含user_id） 2. 从client_user_map缓存获取
                        user_id = None
                        if event == "search" and isinstance(message.get("data"), dict):
                            user_id = message["data"].get("user_id")
                        if not user_id:
                            user_id = ClientManager.get_user_id_by_client_id(client_id) if client_id else None
                        
                        if event == "search" and not user_id:
                            logger.warning(f"[WebSocket] ⚠️ search事件user_id为None: client_id={client_id}")
                        
                        await ClientManager.broadcast_to_websocket({
                            "type": "client_response",
                            "client_id": client_id,
                            "event": event,
                            "message": message
                        }, user_id=user_id)
                else:
                    logger.debug(f"[WebSocket] ⏩ 跳过转发{event}事件（已在处理逻辑中广播）: client_id={client_id}")
            else:
                logger.warning(f"收到响应但无client_id: event={event}, direction={direction}")
            
            return client_id
        
        return None
    
    async def send_message(self, writer: asyncio.StreamWriter, message: dict):
        """发送消息到客户端"""
        try:
            data = (json.dumps(message, ensure_ascii=False) + "\n").encode('utf-8')
            writer.write(data)
            await writer.drain()
        except Exception as e:
            logger.error(f"发送消息失败: {e}")
    
    async def _handle_auto_order(
        self,
        new_items: List[Dict[str, Any]],
        keyword: str,
        order_price_limit: float,
        db_session,
        user_id: Optional[int] = None  # ✅ 新增：用户ID（用于过滤下单客户端）
    ):
        """
        处理自动下单：选择下单客户端执行下单（串行处理，等待每个订单完成后再发送下一个）
        
        Args:
            new_items: 新商品列表
            keyword: 搜索关键词
            order_price_limit: 下单价格限制（0=有新就下，>0=价格<=limit才下）
            db_session: 数据库会话
            user_id: 用户ID（用于过滤下单客户端，确保只使用分配给该用户的下单客户端）
        """
        try:
            # 过滤符合价格条件的新商品
            candidates = []
            for item in new_items:
                price = float(item.get('price', 0) or 0)
                if order_price_limit == 0 or (price > 0 and price <= order_price_limit):
                    candidates.append(item)
            
            if not candidates:
                logger.debug(f"自动下单：新商品价格不符合条件（限制={order_price_limit}），跳过")
                return
            
            # 选择可用的下单客户端
            from config.get_db import get_db
            from module_admin.service.agent_client_service import AgentClientService
            
            async for db in get_db():
                # ✅ 允许使用搜索账号下单（临时方案，用于测试）
                # ✅ 按user_id过滤下单客户端（确保只使用分配给该用户的下单客户端）
                order_clients = await AgentClientService.get_available_order_clients(
                    db, 
                    allow_search_accounts=True,  # 临时允许搜索账号下单
                    user_id=user_id  # ✅ 按user_id过滤
                )
                # ✅ 只选择在线、有默认地址、且状态为idle的客户端
                available = [
                    c for c in order_clients
                    if c.online_status == 'online' 
                    and (c.default_address_id or '').strip()
                    and (c.work_status or 'idle') == 'idle'  # ✅ 只选择idle状态的客户端
                ]
                
                if not available:
                    logger.warning("自动下单：没有可用的下单客户端（在线、有默认地址、且状态为idle）")
                    break
                
                logger.info(f"自动下单：找到 {len(available)} 个可用客户端，准备并发下单 {len(candidates)} 个商品")
                
                # ✅ 获取调度器状态（用于检查任务是否停止）
                from module_task.search_scheduler import scheduler_manager
                scheduler = await scheduler_manager.get_scheduler(user_id) if user_id else None
                
                # ✅ 批量检查pending订单（优化：一次查询所有商品）
                from module_admin.dao.agent_client_dao import AgentClientDao
                from module_admin.dao.search_task_dao import OrderRecordDao
                from sqlalchemy import select
                from module_admin.entity.do.order_record_do import OrderRecord
                
                # 获取所有候选商品ID
                candidate_item_ids = [item.get('item_id') or item.get('id') or item.get('itemId') for item in candidates]
                candidate_item_ids = [str(iid) for iid in candidate_item_ids if iid]
                
                # ✅ 批量查询pending订单（只查询1次数据库）
                pending_item_ids = set()
                if candidate_item_ids:
                    query = select(OrderRecord.item_id).where(
                        OrderRecord.item_id.in_(candidate_item_ids),
                        OrderRecord.order_status == 'pending',
                        OrderRecord.del_flag == '0'
                    )
                    result = await db.execute(query)
                    pending_item_ids = set(row[0] for row in result.fetchall())
                    if pending_item_ids:
                        logger.info(f"自动下单：{len(pending_item_ids)}个商品已有pending订单，跳过: {pending_item_ids}")
                
                # ✅ 过滤掉已有pending订单的商品
                filtered_candidates = [
                    item for item in candidates
                    if str(item.get('item_id') or item.get('id') or item.get('itemId')) not in pending_item_ids
                ]
                
                if not filtered_candidates:
                    logger.info("自动下单：所有商品都已有pending订单，跳过")
                    break
                
                logger.info(f"自动下单：过滤后剩余 {len(filtered_candidates)} 个商品需要下单")
                
                # ✅ 方案：依赖agent端队列机制，后端直接发送所有命令
                # agent端会自动串行处理，避免并发冲突
                # 优点：简单高效，agent端可控制下单节奏
                order_tasks = []
                agent_rotation_index = 0
                
                for item in filtered_candidates:
                    item_id = item.get('item_id') or item.get('id') or item.get('itemId')
                    if not item_id:
                        continue
                    
                    # ✅ 为每个商品创建下单任务（agent端会排队处理）
                    task = self._order_single_item(
                        item=item,
                        keyword=keyword,
                        available_agents=available,
                        agent_rotation_index=agent_rotation_index,
                        scheduler=scheduler,
                        user_id=user_id,
                        AgentClientDao=AgentClientDao,
                        skip_pending_check=True  # ✅ 已经批量检查过，跳过单个检查
                    )
                    order_tasks.append(task)
                    # ✅ 轮转索引递增（均匀分配给不同agent）
                    agent_rotation_index = (agent_rotation_index + 1) % len(available)
                
                # ✅ 并发发送所有命令（agent端会排队处理，不会并发执行）
                if order_tasks:
                    results = await asyncio.gather(*order_tasks, return_exceptions=True)
                    success_count = sum(1 for r in results if r is True)
                    logger.info(f"自动下单命令发送完成：成功 {success_count}/{len(order_tasks)} 个商品（agent端会串行处理）")
                break
                
        except Exception as e:
            logger.error(f"自动下单处理失败: {e}", exc_info=True)
    
    async def _order_single_item(
        self,
        item: Dict[str, Any],
        keyword: str,
        available_agents: List,
        agent_rotation_index: int,
        scheduler,
        user_id: Optional[int],
        AgentClientDao,
        skip_pending_check: bool = False  # ✅ 是否跳过pending检查（已经批量检查过）
    ) -> bool:
        """
        为单个商品下单（支持agent异常跳过和任务状态检查）
        
        Args:
            item: 商品信息
            keyword: 搜索关键词
            available_agents: 可用agent列表
            agent_rotation_index: 轮转索引（从哪个agent开始尝试）
            scheduler: 调度器实例（用于检查任务状态）
            user_id: 用户ID
            AgentClientDao: AgentClientDao类（避免重复导入）
        
        Returns:
            bool: 是否成功发送订单
        """
        item_id = item.get('item_id') or item.get('id') or item.get('itemId')
        if not item_id:
            return False
        
        # ✅ 1. 检查是否已有pending订单（防止重复下单）
        from config.get_db import get_db
        from module_admin.dao.search_task_dao import OrderRecordDao
        from module_admin.entity.do.order_record_do import OrderRecord
        
        # ✅ 修复：使用result变量，避免在async for中return导致连接泄漏
        result = False
        async for db in get_db():
            try:
                # ✅ 如果已经批量检查过，跳过单个检查（性能优化）
                if not skip_pending_check:
                    existing_pending = await OrderRecordDao.get_pending_record_by_item_id(db, str(item_id))
                    if existing_pending:
                        logger.debug(f"自动下单：商品 {item_id} 已有pending订单（order_id={existing_pending.id}），跳过")
                        result = False
                        break  # ✅ 使用break而不是return
                
                # ✅ 2. 选择agent（使用轮转索引，不需要重新检查状态）
                # 注意：外层已经过滤了idle状态的agent，这里直接使用即可
                # pending订单记录会作为锁，防止并发冲突
                
                # 检查任务是否停止
                if scheduler and not scheduler.is_running():
                    logger.info(f"任务已停止，取消商品 {item_id} 的下单")
                    result = False
                    break  # ✅ 使用break而不是return
                
                # 直接使用轮转索引选择agent（外层已过滤）
                agent_index = agent_rotation_index % len(available_agents)
                selected_agent = available_agents[agent_index]
                
                # ✅ 2. 检查是否已有pending订单（防止重复下单）
                existing_pending = await OrderRecordDao.get_pending_record_by_item_id(db, str(item_id))
                if existing_pending:
                    logger.warning(f"⚠️ 自动下单：商品{item_id}已有pending订单(order_id={existing_pending.id})，跳过")
                    result = False
                    break  # ✅ 跳过此商品
                
                # ✅ 3. 创建pending订单记录（占位，防止并发）
                pending_order = OrderRecord(
                    item_id=str(item_id),
                    title=item.get('title', ''),
                    price=item.get('price'),
                    client_id=selected_agent.client_id,
                    order_status='pending',
                    keyword=keyword,
                    seller_id=item.get('seller_id', ''),
                    is_auto_order=True,  # ✅ 标记：自动下单
                    create_by=user_id or 1,
                    dept_id=1
                )
                await OrderRecordDao.create_record(db, pending_order)
                await db.commit()
                logger.info(f"✅ 自动下单：已创建pending订单 item_id={item_id}, order_id={pending_order.id}, client_id={selected_agent.client_id}")
                
                # ✅ 4. 发送下单命令
                order_cmd = {
                    "event": "order",
                    "direction": "req",
                    "success": True,
                    "code": 0,
                    "msg": "",
                    "data": {
                        "item_id": str(item_id),
                        "address_id": selected_agent.default_address_id,
                        "keyword": keyword
                    }
                }
                
                logger.info(f"自动下单：商品 {item_id} 发送到 {selected_agent.client_id}")
                success = await ClientManager.send_to_client(selected_agent.client_id, order_cmd)
                
                if success:
                    result = True
                    break  # ✅ 使用break而不是return
                else:
                    # 发送失败，删除pending订单
                    await OrderRecordDao.delete_record(db, pending_order.id)
                    await db.commit()
                    logger.error(f"❌ 自动下单：发送命令失败，已删除pending订单 order_id={pending_order.id}")
                    result = False
                    break  # ✅ 使用break而不是return
                    
            except Exception as e:
                await db.rollback()
                logger.error(f"自动下单异常: item_id={item_id}, error={e}", exc_info=True)
                result = False
                break  # ✅ 使用break而不是return
        
        return result  # ✅ 在async for之外return
    
    async def start_server(self):
        """启动TCP服务器"""
        self.server = await asyncio.start_server(
            self.handle_client, self.host, self.port
        )
        logger.info(f"TCP服务器已启动: {self.host}:{self.port}")
        
        async with self.server:
            await self.server.serve_forever()
    
    async def stop_server(self):
        """停止TCP服务器"""
        if self.server:
            self.server.close()
            await self.server.wait_closed()
            logger.info("TCP服务器已停止")


# 全局TCP服务器实例
tcp_server = TCPServer()